package com.bml.architect.base;

import com.bml.architect.utils.MqConstants;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

/**
 * 拉模式消费，可自己管理消费的offset
 */
public class LitePullConsumerAssign {

    public static void main(String[] args) {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("consumerPushGrpoup");

        try {
            consumer.setNamesrvAddr(MqConstants.getNameServerAddr());
            consumer.setAutoCommit(true);
            consumer.start();
            Collection<MessageQueue> topicTest = consumer.fetchMessageQueues("TopicTest");
            List<MessageQueue> list = new ArrayList<>(topicTest);
            consumer.assign(list);
            //从队列的第一个开始取，每次从第一个位置开始取消息
            consumer.seek(list.get(0), 1);
            while (true) {
                List<MessageExt> poll = consumer.poll();
                System.out.printf("%s %n", poll);
                //提交已经取的offset，下次从新的offset开始取消息
                consumer.commitSync();
            }

        } catch (MQClientException e) {
            e.printStackTrace();
        } finally {
            consumer.shutdown();
        }

    }
}
